use crate::cmp::Cmp;
use crate::types::{current_key_val, Direction, LdbIterator};

use bytes::Bytes;
use std::cmp::Ordering;
use std::rc::Rc;

// Warning: This module is kinda messy. The original implementation is
// not that much better though :-)
//
// Issues: 1) prev() may not work correctly at the beginning of a merging
// iterator.

#[derive(PartialEq)]
enum SL {
    Smallest,
    Largest,
}

/// 这个 MergingIter 结构体是 LSM 树存储引擎中的合并迭代器，用于将多个底层数据源（如内存表、不同层级的 SSTable）的迭代器结果按键的顺序合并，
/// 向上层提供一个统一的、全局有序的键值对遍历接口。它解决了 LSM 树数据分布在多个独立存储结构中时的有序遍历问题。
pub struct MergingIter {
    /// 存储多个底层迭代器的集合，每个迭代器对应一个数据源（如内存表 memtable、某个层级的 SSTable 文件）；
    /// Box<dyn LdbIterator> 表示 “实现了 LdbIterator trait 的任意类型迭代器”，通过动态分发支持不同类型的数据源（内存或磁盘）。
    iters: Vec<Box<dyn LdbIterator>>,
    /// 记录当前 “最小键 / 最大键” 所在的底层迭代器索引（usize 是迭代器在 iters 中的位置）；
    /// Option 表示可能无有效迭代器（如所有数据源都已遍历完毕）；
    /// 作用：快速定位当前需要处理的底层迭代器，避免每次都扫描所有迭代器。
    current: Option<usize>,
    /// 迭代方向，通常是 Forward（正向，从最小键到最大键）或 Reverse（反向，从最大键到最小键）；
    /// 控制合并逻辑：正向时找所有迭代器中的最小键，反向时找最大键。
    direction: Direction,
    /// 键比较器，定义键的排序规则（如字典序）；
    /// 所有迭代器的键比较都依赖此 comparator，确保全局排序一致。
    cmp: Rc<Box<dyn Cmp>>,
}

impl MergingIter {
    /// Construct a new merging iterator.
    pub fn new(cmp: Rc<Box<dyn Cmp>>, iters: Vec<Box<dyn LdbIterator>>) -> MergingIter {
        MergingIter {
            iters,
            current: None,
            direction: Direction::Forward,
            cmp,
        }
    }

    /// 这段代码定义了一个 init 方法，用于初始化一个包含多个子迭代器的复合迭代器（很可能是之前提到的 MergingIter 或类似的合并迭代器），使其做好开始遍历的准备。
    ///
    /// 确保所有子数据源（如不同的 SSTable 或内存表）的迭代器都处于就绪状态
    ///
    /// 通过 find_smallest() 确定所有数据源中的第一个元素（最小键），为后续的有序合并遍历奠定基础
    fn init(&mut self) {
        println!("----------merge init----------");
        // 遍历 self.iters 中的所有子迭代器（i in 0..self.iters.len()）
        for i in 0..self.iters.len() {
            // 对每个子迭代器执行 reset() 方法，将其重置到初始状态
            self.iters[i].reset();
            // 调用 advance() 方法让子迭代器前进到第一个有效元素的位置
            self.iters[i].advance();
            // 如果子迭代器重置后无效（!valid()），则再次调用 reset() 确保其处于正确的初始状态
            if !self.iters[i].valid() {
                self.iters[i].reset()
            }
        }
        // 调用 self.find_smallest() 从所有子迭代器的当前位置中，找到键值最小的那个元素位置
        // 这一步是合并迭代器的核心准备工作，确保初始化后迭代器指向所有数据源中的第一个元素
        self.find_smallest();
    }

    /// Adjusts the direction of the iterator depending on whether the last
    /// call was next() or prev(). This basically sets all iterators to one
    /// entry after (Forward) or one entry before (Reverse) the current() entry.
    ///
    /// 这个 update_direction 方法是合并迭代器（MergingIter）中用于处理迭代方向切换的核心逻辑。
    /// 当迭代器需要从正向（Forward，从第一个键到最后一个）切换到反向（Reverse，从最后一个键到第一个）或反之，
    /// 该方法会调整所有底层迭代器的状态，确保切换后能正确继续遍历，维持键的有序性。
    ///
    /// 用户先正向遍历
    ///
    /// let mut iter = merging_iter;
    ///
    /// iter.next();  // 正向推进，direction = Forward
    ///
    ///
    /// 切换到反向遍历
    ///
    /// iter.prev();  // 触发 update_direction(Reverse)
    fn update_direction(&mut self, d: Direction) {
        println!("----------merge update_direction----------");
        // 避免无意义的重复操作，只有方向发生变化时才执行后续逻辑。
        if self.direction == d {
            return;  // 若当前方向与目标方向一致，无需调整
        }
        // current_key_val(self) 获取当前合并迭代器指向的键（即所有底层迭代器中的最小 / 最大键）；
        // self.current 是该键所在的底层迭代器索引，用于区分 “当前活跃迭代器” 和其他迭代器。
        if let Some((key, _)) = current_key_val(self) {  // 获取当前迭代器指向的键
            if let Some(current) = self.current {  // 当前最小/最大键所在的底层迭代器索引
                match d {
                    // 场景：之前在反向遍历（从后往前），现在要切换到正向遍历（从前往后）；
                    // 核心操作：
                    // 对所有非当前迭代器执行 seek(&key)，定位到当前键的位置；
                    // 若某迭代器的当前键与目标键完全相同（因 LSM 树中同一键可能有多个版本，通过序列号区分），则推进该迭代器（advance），避免切换后重复返回该键；
                    // 确保切换后，所有迭代器的状态与正向遍历逻辑一致（从当前键之后继续向前）。
                    Direction::Forward if self.direction == Direction::Reverse => {
                        self.direction = Direction::Forward;  // 更新方向为正向
                        // 遍历所有底层迭代器，调整非当前迭代器的状态
                        for i in 0..self.iters.len() {
                            if i != current {  // 跳过当前活跃迭代器（已由 advance/prev 处理）
                                self.iters[i].seek(&key);  // 定位到当前键的位置
                                // This doesn't work if two iterators are returning the exact same
                                // keys. However, in reality, two entries will always have differing
                                // sequence numbers.
                                // 若该迭代器的当前键与目标键相等，推进到下一个键（避免重复）
                                if let Some((current_key, _)) = self.iters[i].current() {
                                    if self.cmp.cmp(&current_key, &key) == Ordering::Equal {
                                        self.iters[i].advance();
                                    }
                                }
                            }
                        }
                    }
                    // 场景：之前在正向遍历（从前往后），现在要切换到反向遍历（从后往前）；
                    // 核心操作：
                    // 对所有非当前迭代器执行 seek(&key)，定位到当前键的位置；
                    // 若迭代器有效（找到了当前键），则回退到上一个键（prev），确保反向遍历时从当前键之前的键开始；
                    // 若迭代器无效（未找到当前键），则推进到最后一个键（while advance()），作为反向遍历的起点；
                    // 确保切换后，所有迭代器的状态与反向遍历逻辑一致（从当前键之前继续向后）。
                    Direction::Reverse if self.direction == Direction::Forward => {
                        self.direction = Direction::Reverse;  // 更新方向为反向
                        // 遍历所有底层迭代器，调整非当前迭代器的状态
                        for i in 0..self.iters.len() {
                            if i != current {  // 跳过当前活跃迭代器
                                self.iters[i].seek(&key);  // 定位到当前键的位置
                                if self.iters[i].valid() {
                                    self.iters[i].prev();  // 若有效，回退到上一个键
                                } else {
                                    // seek to last.
                                    // 若无效（未找到当前键），则推进到最后一个键
                                    while self.iters[i].advance() {}
                                }
                            }
                        }
                    }
                    _ => {}
                }
            }
        }
    }

    fn find_smallest(&mut self) {
        println!("----------merge find_smallest----------");
        self.find(SL::Smallest)
    }
    fn find_largest(&mut self) {
        println!("----------merge find_largest----------");
        self.find(SL::Largest)
    }

    /// 这个 find 方法是合并迭代器（MergingIter）内部用于定位所有底层迭代器中的最小或最大键的核心逻辑。
    /// 它根据指定的方向（找最小键或最大键），从多个底层迭代器中筛选出当前符合条件的那个迭代器，并记录其索引，为后续的有序遍历提供基础。
    fn find(&mut self, direction: SL) {
        println!("----------merge find----------");
        // 当没有任何底层数据源（如内存表和 SSTable 都为空）时，无需处理，直接返回（此时迭代器状态为无效）。
        if self.iters.is_empty() {
            // Iterator stays invalid.
            // 若没有底层迭代器，迭代器保持无效状态
            return;
        }
        // 根据目标方向（Smallest 或 Largest）确定比较逻辑：
        // 找最小键时，若一个键 “小于” 当前候选键，则它更可能是目标；
        // 找最大键时，若一个键 “大于” 当前候选键，则它更可能是目标。
        let ord = if direction == SL::Smallest {
            Ordering::Less  // 找最小键：比较时用 "小于" 规则
        } else {
            Ordering::Greater  // 找最大键：比较时用 "大于" 规则
        };

        let mut next_ix = 0;  // 初始候选迭代器索引（默认第一个）

        // 从第二个迭代器（i=1）开始遍历所有底层迭代器：
        // 对每个迭代器 i，先检查它是否有有效键（current() 返回 Some）；
        // 再检查当前候选迭代器（next_ix）是否有有效键：
        // 若两者都有效，用比较器（self.cmp）比较它们的键，若迭代器 i 的键更符合目标方向（更小或更大），则将 next_ix 更新为 i；
        // 若候选迭代器无效（无键），则直接将 next_ix 更新为 i（因为迭代器 i 有有效键）。
        for i in 1..self.iters.len() {
            // 获取当前迭代器（i）的当前键
            if let Some((current_key, _)) = self.iters[i].current() {
                // 获取候选迭代器（next_ix）的当前键
                if let Some((smallest_key, _)) = self.iters[next_ix].current() {
                    // 比较当前键与候选键，若符合方向规则（小于/大于），则更新候选索引
                    if self.cmp.cmp(&current_key, &smallest_key) == ord {
                        next_ix = i;
                    }
                } else {
                    // 候选迭代器无效（无当前键），直接替换为当前迭代器
                    next_ix = i;
                }
            }
        }
        // 最终 next_ix 指向所有有效迭代器中 “键最小” 或 “键最大” 的那个迭代器，通过 self.current 记录，供后续迭代（如 next、prev）使用。
        self.current = Some(next_ix);  // 将找到的迭代器索引存入 current
    }
}

impl LdbIterator for MergingIter {
    /// 这个 advance 方法是合并迭代器（MergingIter）中用于向前推进到下一个最小键值对的核心逻辑。
    /// 合并迭代器的作用是将多个底层迭代器（如来自不同 SSTable 或内存表的迭代器）的结果按键的顺序合并，而 advance 方法则负责在向前迭代时更新状态，确保始终能获取到当前所有有效迭代器中的最小键。
    fn advance(&mut self) -> bool {
        println!("----------merge advance----------");
        // self.current 记录当前最小键所在的底层迭代器索引（如第 i 个 SSTable 迭代器）；
        // 首先将迭代方向更新为向前（Forward），确保后续操作符合正向遍历逻辑；
        // 推进当前最小键所在的迭代器（self.iters[current].advance()）：
        // 若推进成功（该迭代器还有下一个键），则继续；
        // 若推进失败（该迭代器已遍历完毕），则重置该迭代器（使其被标记为无效，不再参与后续比较）；
        // 调用 self.find_smallest() 重新扫描所有有效迭代器，找到新的最小键，更新 self.current 指向该迭代器。
        if let Some(current) = self.current {
            self.update_direction(Direction::Forward); // 更新迭代方向为向前
            // 推进当前活跃的迭代器（current 是当前最小键所在的迭代器索引）
            if !self.iters[current].advance() {
                // Take this iterator out of rotation; this will return false
                // for every call to current() and thus it will be ignored
                // from here on.
                // 若推进失败（当前迭代器已无更多数据），重置该迭代器（标记为无效）
                self.iters[current].reset();
            }
            // 重新寻找所有有效迭代器中的最小键
            self.find_smallest();
        } else {
            // 当 self.current 为 None 时（如迭代器刚创建时），调用 self.init() 进行初始化：
            // init 方法会启动所有底层迭代器，定位到各自的第一个键；
            // 然后通过 find_smallest 找到所有迭代器中的最小键，设置 self.current 为该迭代器的索引。
            self.init(); // 初始化迭代器（寻找第一个最小键）
        }
        // valid() 方法检查 self.current 是否指向一个有效的迭代器（该迭代器有可用键），返回 true 表示迭代器处于有效状态（可获取下一个键）。
        self.valid() // 返回当前是否有有效键（即是否找到最小键）
    }
    /// 这个 valid 方法用于判断合并迭代器（MergingIter）当前是否处于有效状态，即是否指向一个可以访问的键值对。
    /// 它是迭代器对外提供的核心状态查询接口，帮助调用者判断是否可以安全地获取当前键值对或继续推进迭代器。
    fn valid(&self) -> bool {
        println!("----------merge valid----------");
        // self.current 存储了当前 “最小键 / 最大键” 所在的底层迭代器索引（Option<usize>）；
        // 若 self.current 为 Some(ix)（存在活跃迭代器），则通过 self.iters[ix].valid() 检查该底层迭代器是否处于有效状态（即是否指向一个可访问的键值对）；
        // 若 self.current 为 None（无活跃迭代器，如所有底层迭代器都已遍历完毕），则直接返回 false。
        if let Some(ix) = self.current {
            self.iters[ix].valid()   // 检查当前迭代器是否有效
        } else {
            false // 无活跃迭代器，直接返回无效
        }
    }
    fn seek(&mut self, key: &[u8]) {
        println!("----------merge seek----------");
        for i in 0..self.iters.len() {
            self.iters[i].seek(key);
        }
        self.find_smallest();
    }
    fn reset(&mut self) {
        println!("----------merge reset----------");
        for i in 0..self.iters.len() {
            // 这个进的是 version iter
            self.iters[i].reset();
        }
        self.current = None;
    }
    /// 这个 current 方法是合并迭代器（MergingIter）用于获取当前指向的键值对的核心接口。
    /// 它返回当前所有底层迭代器中 “最小键” 或 “最大键” 对应的键值对（取决于迭代方向），为上层提供合并后的有序数据。
    ///
    /// self.current 存储了当前 “最小键 / 最大键” 所在的底层迭代器索引（Option<usize>）；
    ///
    /// 若存在活跃迭代器（Some(ix)），则调用该底层迭代器的 current() 方法，返回其指向的键值对；
    ///
    /// 若不存在活跃迭代器（None，如所有底层迭代器已遍历完毕），则返回 None。
    fn current(&self) -> Option<(Bytes, Bytes)> {
        println!("----------merge current----------");
        if let Some(ix) = self.current {
            self.iters[ix].current()  // 返回当前迭代器的键值对
        } else {
            None // 无活跃迭代器，返回 None
        }
    }
    fn prev(&mut self) -> bool {
        println!("----------merge prev----------");
        if let Some(current) = self.current {
            if self.iters[current].valid() {
                self.update_direction(Direction::Reverse);
                self.iters[current].prev();
                self.find_largest();
                self.valid()
            } else {
                false
            }
        } else {
            false
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use crate::cmp::DefaultCmp;
    use crate::skipmap::tests;
    use crate::test_util::{test_iterator_properties, LdbIteratorIter, TestLdbIter};
    use crate::types::{current_key_val, LdbIterator};

    #[test]
    fn test_merging_one() {
        let skm = tests::make_skipmap();
        let iter = skm.iter();
        let mut iter2 = skm.iter();

        let mut miter = MergingIter::new(Rc::new(Box::new(DefaultCmp)), vec![Box::new(iter)]);

        while let Some((k, v)) = miter.next() {
            if let Some((k2, v2)) = iter2.next() {
                assert_eq!(k, k2);
                assert_eq!(v, v2);
            } else {
                panic!("Expected element from iter2");
            }
        }
    }

    #[test]
    fn test_merging_two() {
        let skm = tests::make_skipmap();
        let iter = skm.iter();
        let iter2 = skm.iter();

        let mut miter = MergingIter::new(
            Rc::new(Box::new(DefaultCmp)),
            vec![Box::new(iter), Box::new(iter2)],
        );

        while let Some((k, v)) = miter.next() {
            if let Some((k2, v2)) = miter.next() {
                assert_eq!(k, k2);
                assert_eq!(v, v2);
            } else {
                panic!("Odd number of elements");
            }
        }
    }

    #[test]
    fn test_merging_zero() {
        let mut miter = MergingIter::new(Rc::new(Box::new(DefaultCmp)), vec![]);
        assert_eq!(0, LdbIteratorIter::wrap(&mut miter).count());
    }

    #[test]
    fn test_merging_behavior() {
        let val = b"def";
        let iter = TestLdbIter::new(vec![(b("aba"), val), (b("abc"), val)]);
        let iter2 = TestLdbIter::new(vec![(b("abb"), val), (b("abd"), val)]);
        let miter = MergingIter::new(
            Rc::new(Box::new(DefaultCmp)),
            vec![Box::new(iter), Box::new(iter2)],
        );
        test_iterator_properties(miter);
    }

    #[test]
    fn test_merging_forward_backward() {
        let val = b"def";
        let iter = TestLdbIter::new(vec![(b("aba"), val), (b("abc"), val), (b("abe"), val)]);
        let iter2 = TestLdbIter::new(vec![(b("abb"), val), (b("abd"), val)]);

        let mut miter = MergingIter::new(
            Rc::new(Box::new(DefaultCmp)),
            vec![Box::new(iter), Box::new(iter2)],
        );

        // miter should return the following sequence: [aba, abb, abc, abd, abe]

        // -> aba
        let first = miter.next();
        // -> abb
        let second = miter.next();
        // -> abc
        let third = miter.next();
        eprintln!("{:?} {:?} {:?}", first, second, third);

        assert!(first != third);
        // abb <-
        assert!(miter.prev());
        assert_eq!(second, current_key_val(&miter));
        // aba <-
        assert!(miter.prev());
        assert_eq!(first, current_key_val(&miter));
        // -> abb
        assert!(miter.advance());
        assert_eq!(second, current_key_val(&miter));
        // -> abc
        assert!(miter.advance());
        assert_eq!(third, current_key_val(&miter));
        // -> abd
        assert!(miter.advance());
        assert_eq!(
            Some((b("abd").to_vec(), val.to_vec())),
            current_key_val(&miter)
        );
    }

    fn b(s: &'static str) -> &'static [u8] {
        s.as_bytes()
    }

    #[test]
    fn test_merging_real() {
        let val = b"def";

        let it1 = TestLdbIter::new(vec![(b("aba"), val), (b("abc"), val), (b("abe"), val)]);
        let it2 = TestLdbIter::new(vec![(b("abb"), val), (b("abd"), val)]);
        let expected = [b("aba"), b("abb"), b("abc"), b("abd"), b("abe")];

        let mut iter = MergingIter::new(
            Rc::new(Box::new(DefaultCmp)),
            vec![Box::new(it1), Box::new(it2)],
        );

        for (i, (k, _)) in LdbIteratorIter::wrap(&mut iter).enumerate() {
            assert_eq!(k, expected[i]);
        }
    }

    #[test]
    fn test_merging_seek_reset() {
        let val = b"def";

        let it1 = TestLdbIter::new(vec![(b("aba"), val), (b("abc"), val), (b("abe"), val)]);
        let it2 = TestLdbIter::new(vec![(b("abb"), val), (b("abd"), val)]);

        let mut iter = MergingIter::new(
            Rc::new(Box::new(DefaultCmp)),
            vec![Box::new(it1), Box::new(it2)],
        );

        assert!(!iter.valid());
        iter.advance();
        assert!(iter.valid());
        assert!(current_key_val(&iter).is_some());

        iter.seek(b"abc");
        assert_eq!(
            current_key_val(&iter),
            Some((b("abc").to_vec(), val.to_vec()))
        );
        iter.seek(b"ab0");
        assert_eq!(
            current_key_val(&iter),
            Some((b("aba").to_vec(), val.to_vec()))
        );
        iter.seek(b"abx");
        assert_eq!(current_key_val(&iter), None);

        iter.reset();
        assert!(!iter.valid());
        iter.next();
        assert_eq!(
            current_key_val(&iter),
            Some((b("aba").to_vec(), val.to_vec()))
        );
    }
}
